package com.qupeng.demo.kafka.kafkaspringbootstreams.timewindow;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.utils.TimeUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class TimeWindowTestingOutputListener {

    @KafkaListener(id = "time-window-testing-listener",
            groupId = "timeWindowTestingGroup",
            topics = "time-window-testing-output",
            clientIdPrefix = "time-window-testing",
            properties = {"value.deserializer=org.apache.kafka.common.serialization.LongDeserializer"}
    )
    public void listen(ConsumerRecord<String, Long> consumerRecord) {
        System.out.println(consumerRecord);
        System.out.println(consumerRecord.key());
        System.out.println(consumerRecord.value());
        System.out.println(TimeUtils.format(consumerRecord.timestamp()));
    }
}
